package cdrprocessing.bolt.util;

import java.util.Map;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;

public abstract class WindowAwareBolt extends BaseRichBolt {

    private int counter;

    private int windowSize;

    protected OutputCollector collector;

    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        counter = 0;
        windowSize = Integer.valueOf((String) stormConf.get("windowSize"));

        this.collector = collector;
    }

    public void execute(Tuple input) {
        if (counter == 0) {
            handleStartWindow();
        }

        counter++;
        handleTuple(input);

        if (counter == windowSize) {
            handleEndWindow();
            counter = 0;
        }
    }

    protected abstract void handleStartWindow();

    protected abstract void handleEndWindow();

    protected abstract void handleTuple(Tuple input);
}
